package org.itew.mymapreduce.core.impl;

import java.io.IOException;
import java.util.Collection;
import java.util.Iterator;
import java.util.Set;
import java.util.TreeSet;

import org.itew.mymapreduce.core.DataCollection;
import org.itew.mymapreduce.core.Partition;
import org.itew.mymapreduce.util.PropertiesConfigurationReader;

public class SortedPartition<K, V> implements Partition<K, V> {

	private Iterator<K>[] itetators;
	@SuppressWarnings("rawtypes")
	private Comparable[] keys;
	private int keySetSize;
	
	@SuppressWarnings({ "unchecked", "rawtypes" })
	@Override
	public void setData(Collection<DataCollection<K, V>> mappedDataCollections,
			int reducerNumber) throws IOException {

		PropertiesConfigurationReader pcr = new PropertiesConfigurationReader(SortedPartition.class, "sortedpartition.properties");
		
		keySetSize = pcr.getInt("keySetSize");
		
		int i = 0;
		itetators = new Iterator[mappedDataCollections.size()];
		keys = new Comparable[mappedDataCollections.size()];
		
		
		for (DataCollection<K, V> mappedDataCollection : mappedDataCollections) {
			itetators[i++] = mappedDataCollection.keySet().iterator();
		}
		
		for(i=0;i<itetators.length;i++){
			if(itetators[i].hasNext())
				keys[i] = (Comparable) itetators[i].next();
		}
		
	}

	@SuppressWarnings({ "unchecked", "rawtypes" })
	@Override
	public Set<K> nextKeySet() {
		
		Set<K> keySet = new TreeSet<K>();

		for(int currentKeySetSize=0;currentKeySetSize<keySetSize;currentKeySetSize++){
			
			boolean isFirst = true;
			Comparable key = null;
			int keyIndex = -1;
			
			
			for(int i =0;i<keys.length;i++){
				if(isFirst&&keys[i]!=null){
					isFirst=false;
					key = keys[i];
					keyIndex = i;
				}else if(keys[i]!=null){
					if(key.compareTo(keys[i])==0){
						if(itetators[i].hasNext())
							keys[i] = (Comparable) itetators[i].next();
						else
							keys[i] = null;
					}else if(key.compareTo(keys[i])>0){
						key = keys[i];
						keyIndex = i;
					}
				}
			}
			
			if(key==null){
				if(keySet.isEmpty())
					return null;
				else
					return keySet;
			}
			
			if(itetators[keyIndex].hasNext())
				keys[keyIndex] = (Comparable) itetators[keyIndex].next();
			else
				keys[keyIndex] = null;
			keySet.add((K) key);
			
			
		}
		
		
		
		return keySet;
	}

}
